package com.atguigu.gmall.service.config;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.common.entity.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.UUID;

/**
 * @author ：MJS
 * @date ：Created in 2021/5/26  20:00
 * @description：TODO MQ
 * producer:生产者
 * ack：应答
 * config：配置类
 * 保证生产者不丢失数据
 * 发送确认
 */
@Component//被扫描
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    /*
    *先走private RabbitTemplate rabbitTemplate;
    * 再 @PostConstruct //初始化注解
     public void MQProducerAckConfig()
    * */
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct //初始化注解
    public void MQProducerAckConfig() {
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
    }

    //交换机应答方法
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //ack true 应答成功  false：应答失败
        if (ack) {
            log.info("交换机应答成功:" + cause);
        } else {
            log.info("交换机应答失败:" + cause);
            //重新发送刚才的消息
            retryMessage(correlationData);
        }
    }

    //队列应答方法  只有失败才应答
    @Override
    public void returnedMessage(Message message,
                                int replyCode,
                                String replyText,
                                String exchange,
                                String routingKey) {
        log.info("1队列应答失败:" + exchange + ":" + routingKey);
        log.info("2队列应答失败:" + replyCode + ":" + replyText);
        //判断是基于插件的延迟消息  不要重新发送 (可能出现消息真的丢失   且不在重新发送)



        //没有次数的话  会一直重发  所以不能像交换机那样
        //借助外力 发送 记录次数
        String cacheKey = "mq:retry:message";
        Object uuid = message.getMessageProperties().
                getHeader("spring_returned_message_correlation");
        String json=(String)redisTemplate.
                opsForHash().get(cacheKey,uuid);
        GmallCorrelationData correlationData = JSONObject.parseObject(json, GmallCorrelationData.class);
        //判断是基于插件的延迟消息  不要重新发送 (可能出现消息真的丢失   且不在重新发送)
        if(correlationData.isChaJian()){
            System.out.println("基于插件的暂时就不发了  ");
            return;
        }
        //重新发送
        retryMessage(correlationData);
    }

    //重新发送消息

    private void retryMessage(CorrelationData correlationData) {
        GmallCorrelationData correlationData1 = (GmallCorrelationData) correlationData;
        //重新发送3次  如果三次之后还是不成功  写之日记录
        int retryCount = correlationData1.getRetryCount();
        if (retryCount < 3) {
            retryCount++;
            correlationData1.setRetryCount(retryCount);
            log.info("重新发送第" + retryCount + "次");
            //更新redis缓存中得重新发送的次数
            String cacheKey = "mq:retry:message";
            //相当于删除并更新了
            redisTemplate.opsForHash().put(cacheKey,correlationData1.getId(), JSONObject.toJSONString(correlationData));
            //重新发送
            rabbitTemplate.convertAndSend(correlationData1.getExchange(),
                    correlationData1.getRoutingKey(),
                    correlationData1.getMessage(),
                    correlationData1);
        } else {
            log.error("重新发送消息3次，仍然失败：" + JSONObject.toJSONString(correlationData1));
        }
    }
}
